package com.fengkai.filepost.pdfoutstream.test.common.mq.producer;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

/**
 * @author Fengkai
 * @creat 2021/12/28
 */
public class OrderProducer {
    //同步生产者
       public static void main(String[] args) throws Exception {
           DefaultMQProducer pg = new DefaultMQProducer("pg");
           pg.setNamesrvAddr("192.168.157.130:9876");
           pg.setRetryTimesWhenSendFailed(3);
           //若为全局有序则需要指定queue数量
//           pg.setDefaultTopicQueueNums(1);
           pg.start();

           for (int i = 0; i <5000 ; i++) {
               Integer orderId = i;
               byte[] body = ("HI,TopicOrder" + i).getBytes();
               Message message = new Message("TopicOrder", "TagOrder", body);
               //为消息指定key 将orderId作为消息key
               message.setKeys(orderId.toString() );
               //send 第三个参数将作为选择key参数值传递
               SendResult sendResult = pg.send(message, new MessageQueueSelector() {
                   @Override
                   public MessageQueue select (List<MessageQueue> list, Message message, Object o) {
                       //使用o作为选择key
//                       Integer id = (Integer) o;
//                       int index = id % list.size();

                       //使用消息key作为选择key
                       String keys = message.getKeys();
                       Integer id = Integer.valueOf(keys);
                       int index = id % list.size();
                       return list.get(index);

                   }
               }, orderId);
               System.out.println(sendResult);
           }
            pg.shutdown();
       }
}
